#include "myrpcprovider.h"
#include "myrpcapplication.h"
#include "myrpcheader.pb.h"
#include "myrpclogger.h"

void MyRPCProvider::NotifyService(google::protobuf::Service* service) {
    // 服务信息
    ServiceInfo serviceInfo;

    // 获取服务描述器指针
    const google::protobuf::ServiceDescriptor* psd = service->GetDescriptor();

    // 获取服务的名字
    std::string service_name = psd->name();

    // 获取该服务对应的方法数量
    int method_count = psd->method_count();

    std::cout << "service name:" << service_name << "\n";

    for (int i = 0; i < method_count; ++i) {
        // 获取方法描述器指针
        const google::protobuf::MethodDescriptor* pmd = psd->method(i);
        // 获取方法名
        std::string method_name = pmd->name();
        std::cout << "method name:" << method_name << "\n";
        serviceInfo.method_map.emplace(method_name,pmd);
    }
    serviceInfo.service = service;
    service_map.emplace(service_name, serviceInfo);
}

void MyRPCProvider::Run() {
    // 获取rpc服务的ip
    std::string ip = MyRPCApplication::getInstance().getConfig().load("rpcserverip");
    int port = atoi(MyRPCApplication::getInstance().getConfig().load("rpcserverport").c_str());

    muduo::net::InetAddress address(ip,port);

    std::shared_ptr<muduo::net::TcpServer> server = std::make_shared<muduo::net::TcpServer>(&eventLoop,address,"MyRPCProvider");

    // 设置新连接回调函数
    server->setConnectionCallback(std::bind(&MyRPCProvider::OnConnection, this, std::placeholders::_1));

    // 设置消息到达回调函数
    server->setMessageCallback(std::bind(&MyRPCProvider::OnMessage, this, std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));

    // 设置线程数
    server->setThreadNum(4);

    // 服务端这里与zookeeper连接是为了接下来将服务对象和方法对象
    // 以及对应的ip和port注册到zookeeper注册中心
    ZKClient zkclient;
    zkclient.Start();

    // 开始遍历服务对象map
    for (auto &sp : service_map) {
        // 记录服务对象的名字
        std::string service_path = "/" + sp.first;
        // 创建出服务对象永久性结点
        zkclient.Create(service_path.c_str(),nullptr,0);

        // 然后遍历服务对象的方法
        for (auto &mp : sp.second.method_map) {
            // 获取方法名
            std::string method_path = service_path + "/" + mp.first;

            char method_path_data[128] = {0};
            // 格式化ip和端口
            sprintf(method_path_data,"%s:%d",ip.c_str(),port);
            // 创建方法对象临时性结点，临时结点中保存了ip和端口
            zkclient.Create(method_path.c_str(),method_path_data, strlen(method_path_data),ZOO_EPHEMERAL);
        }
    }
    // RPC服务器启动
    server->start();
    // 事件监听开始
    eventLoop.loop();
}

void MyRPCProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn) {
    if (!conn->connected()) {
        conn->shutdown();
    }
}

void MyRPCProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn, 
                            muduo::net::Buffer* buffer, 
                            muduo::Timestamp receive_time) 
{
    std::cout << "OnMessage \n";
    // 从网络上接收来自客户端发送的数据
    std::string recv_buf = buffer->retrieveAllAsString();

    // 下面开始解析发送过来的这部分数据
    google::protobuf::io::ArrayInputStream raw_input(recv_buf.data(), recv_buf.size());
    google::protobuf::io::CodedInputStream code_input(&raw_input);

    uint32_t header_size = 0;
    // 1、解析出头部字段大小数据
    code_input.ReadVarint32(&header_size);

    // 定义变量保存解析出来的数据
    std::string rpc_header_data;
    
    myrpc::RpcHeader rpcheader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size = 0;
    // 设置读取限制
    google::protobuf::io::CodedInputStream::Limit msg_limit = code_input.PushLimit(header_size);
    // 2、解析出头部的三个数据
    code_input.ReadString(&rpc_header_data, header_size);
    code_input.PopLimit(msg_limit);
    // 将数据解析到rpcheader中
    if (rpcheader.ParseFromString(rpc_header_data)) {
        // 分别获取服务对象名，方法名以及方法参数的字节数
        service_name = rpcheader.service_name();
        method_name = rpcheader.method_name();
        args_size = rpcheader.args_size();
    } else {
        // 反序列化失败
        std::cout << "rpc header_data " << rpc_header_data << " parse error " << std::endl;
        return;
    }

    // 客户端传递的参数字符流
    std::string args_str;
    // 3、解析出参数数据
    bool read_args_str = code_input.ReadString(&args_str, args_size);
    if (!read_args_str) {
        // 参数数据解析失败
        MyRPCLooger::ERROR("read args failed");
    }
    // 打印调试信息
    // std::cout << "===========================================\n";
    // std::cout << "header_size: " << header_size << "\n";
    // std::cout << "rpc_header_data: " << rpc_header_data << "\n";
    // std::cout << "service_name: " << service_name << "\n";
    // std::cout << "method_name: " << method_name << "\n";
    // std::cout << "args_str: " << args_str << "\n";
    // std::cout << "===========================================\n";

    // 通过服务对象名和方法名寻找对应的迭代器
    auto it = service_map.find(service_name);
    if (it == service_map.end()) {
        std::cout << service_name <<" is not exits\n";
        return; 
    }

    auto mit = it->second.method_map.find(method_name);
    if (mit == it->second.method_map.end()) {
        std::cout << method_name << " is not exits\n";
        return;
    }

    // 分别获取服务对象和方法对象
    google::protobuf::Service* service = it->second.service;
    const google::protobuf::MethodDescriptor* method = mit->second;

    google::protobuf::Message* request = service->GetRequestPrototype(method).New();
    if (!request->ParseFromString(args_str)) {
        std::cout << service_name << " " << method_name << " parse error \n";
        return;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();

    // 定义一个Closure回调函数，该函数会在rpc远端调用方法被调用后再调用
    google::protobuf::Closure* done = google::protobuf::NewCallback<MyRPCProvider,
                                                                    const muduo::net::TcpConnectionPtr &,
                                                                    google::protobuf::Message*>(this,
                                                                    &MyRPCProvider::SendRpcMessage,conn,response);
    // 开始调用服务端上的远端客户端请求的方法
    // CallMethod方法内部会通过method参数拿到对应方法的index值，根据index值
    // 调用注册的对应的方法
    service->CallMethod(method,nullptr,request,response,done);
}

void MyRPCProvider::SendRpcMessage(const muduo::net::TcpConnectionPtr& conn, 
                                    google::protobuf::Message* response) 
{
    // 开始序列化为字符串
    std::string response_str;
    if (response->SerializeToString(&response_str)) {
        conn->send(response_str);
    } else {
        std::cout << "serialize error \n";
    }
    // 断开连接
    conn->shutdown();
}

MyRPCProvider::~MyRPCProvider() {
    std::cout << "~MyRPCProvider() \n";
    eventLoop.quit();
}